package org.appchat.rabbitmq;

import com.google.common.base.Throwables;
import com.google.common.collect.Maps;
import com.google.common.collect.Sets;
import com.google.protobuf.*;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.*;
import org.appchat.api.Listener;
import org.appchat.api.Publisher;

//import javax.jms.*;
//import javax.jms.Connection;
//import javax.jms.ConnectionFactory;
//import javax.jms.Message;
import java.io.IOException;
import java.util.Map;
import java.util.Set;
import java.io.IOException;
import com.rabbitmq.client.ConnectionFactory;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.QueueingConsumer;

public class ChatRoom {
    private static final String EXCHANGE_NAME = "topic_logs";

    private final String brokerUrl;
    private final Set<Listener> subscriptions;

    //private final Map<String, Destination> topicNamesToTopics = Maps.newHashMap();
    //private final Map<String, MessageProducer> queueNamesToProducers = Maps.newHashMap();

    //private Session session;
    private PublisherImpl publisher;
    private Receiver receiver;

    private com.rabbitmq.client.ConnectionFactory connectionFactory;

    private ChatRoom(Builder builder) {
        subscriptions = builder.subscriptions;

        if (builder.brokerUrl == null) {
            brokerUrl = com.rabbitmq.client.ConnectionFactory.DEFAULT_HOST;//ActiveMQConnectionFactory.DEFAULT_BROKER_URL;
        } else {
            brokerUrl = builder.brokerUrl;
        }
    }

    public void startChat() {
        QueueingConsumer consumer = null;

        try{
            ConnectionFactory factory = new ConnectionFactory();
            factory.setHost("localhost");
            Connection connection = factory.newConnection();
            Channel channel = connection.createChannel();

            channel.exchangeDeclare(EXCHANGE_NAME, "fanout");
            String queueName = channel.queueDeclare().getQueue();
            channel.queueBind(queueName, EXCHANGE_NAME, "");

            System.out.println(" [*] Waiting for messages. To exit press CTRL+C");

            consumer = new QueueingConsumer(channel);
            channel.basicConsume(queueName, true, consumer);
        }catch (IOException e){throw Throwables.propagate(e);}


        receiver = new Receiver(consumer);
        publisher = new PublisherImpl();
    }

    public void stopChat() {
        //TODO Implement this
    }

    public Publisher getPublisher() {
        return publisher;
    }

    public static Builder builder() {
        return new Builder();
    }

    public class PublisherImpl implements Publisher {

        @Override
        public void publish(Message message) {
            String messageDescriptor = message.getDescriptorForType().getFullName();


            try {

                ConnectionFactory factory = new ConnectionFactory();
                factory.setHost("localhost");
                Connection connection = factory.newConnection();
                Channel channel = connection.createChannel();

                channel.exchangeDeclare(EXCHANGE_NAME, "fanout");

                //String message = getMessage(argv);

                channel.basicPublish(EXCHANGE_NAME, "", null, message.toByteArray());
                System.out.println(" [x] Sent '" + message + "'");

                channel.close();
                connection.close();

            } catch (IOException e) {
                throw Throwables.propagate(e);
            }
        }
    }

    public static class Builder {

        private String brokerUrl;
        Set<Listener> subscriptions = Sets.newHashSet();

        public Builder setBrokerUrl(String brokerUrl) {
            this.brokerUrl = brokerUrl;
            return this;
        }

        public Builder listenFor(Listener listener) {
            this.subscriptions.add(listener);
            return this;
        }

        public ChatRoom build() {
            return new ChatRoom(this);
        }
    }


    //playing with this for multithreaded polling.
    private class Receiver implements Runnable {
        private final QueueingConsumer consumer;
        public Receiver(QueueingConsumer consumer){
            this.consumer = consumer;
            Thread t = new Thread(this, "Receiver");
            t.start();
        }

        @Override
        public void run() {
            String message = "message has not been initialized";
            //while (true) {
                try{
                QueueingConsumer.Delivery delivery = consumer.nextDelivery();
                message = new String(delivery.getBody());
                } catch(InterruptedException e){e.printStackTrace();}

                System.out.println(" [x] Received '" + message + "'");
            //}
        }
    }
}
